package org.jetlinks.pro.things.impl.converter;

import com.google.common.collect.Maps;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.function.ThingFunctionInvokeMessage;
import org.jetlinks.core.message.function.ThingFunctionInvokeMessageReply;
import org.jetlinks.core.message.property.*;
import org.jetlinks.pro.things.ThingMessageConverter;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Mono;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 物消息转换器,将消息模版和上下文转为新的消息
 *
 * @author zhouhao
 * @since 1.12
 */
@Getter
@Setter
public class AutoThingMessageConverter implements ThingMessageConverter {

    /**
     * 功能调用返回单个值的key
     */
    public static final String functionOutputKey = "__response";

    /**
     * 转换消息时需要排除的标识
     */
    private Set<String> excludes;

    /**
     * 转换消息时需要必须要包含的标识。当includes不为空，并且正在处理标识没在includes中则忽略转换不作处理
     */
    private Set<String> includes;


    /**
     * 转换消息
     *
     * @param context  上下文
     * @param template 模版
     * @return 新的消息
     */
    @Override
    public Mono<ThingMessage> convert(Map<String, Object> context, ThingMessage template) {
        return doConvert(getNeedConvertInfo(template), context, template.copy());
    }

    /**
     * 做消息转换
     *
     * @param expr 物消息中需要转换的信息
     * @param context 上下文
     * @param message 消息模版
     * @return 新的消息
     */
    private Mono<ThingMessage> doConvert(Map<String, Object> expr, Map<String, Object> context, ThingMessage message) {
        Map<String, Object> parsed = doConvert(expr, context);
        //判断物消息为属性上报，将转换结果作为属性上报的参数
        if (message instanceof ThingReportPropertyMessage) {
            //没有解析出属性上报，表示当前属性上报的的所有信息都没有处理意义，所以返回空流
            if (MapUtils.isEmpty(parsed)) {
                return Mono.empty();
            }
            ThingReportPropertyMessage msg = ((ThingReportPropertyMessage) message);
            msg.success(parsed);
            return Mono.just(msg);
            //判断物消息为功能回复，将转换结果作为功能回复的输出参数
        } else if (message instanceof ThingFunctionInvokeMessageReply) {
            ThingFunctionInvokeMessageReply msg = ((ThingFunctionInvokeMessageReply) message);
            msg.success();
            Object output = parsed.get(functionOutputKey);
            if (null != output) {
                msg.output(output);
            } else {
                msg.output(parsed);
            }
            return Mono.just(msg);
            //判断物消息为功能调用，将转换结果作为功能调用的入参
        } else if (message instanceof ThingFunctionInvokeMessage) {
            ThingFunctionInvokeMessage<?> msg = ((ThingFunctionInvokeMessage<?>) message);
            parsed.forEach(msg::addInput);
            return Mono.just(msg);
            //判断物消息为读取属性消息，将转换结果集合key作为需要读取属性的集合
        } else if (message instanceof ReadThingPropertyMessage) {
            ReadThingPropertyMessage<?> msg = ((ReadThingPropertyMessage<?>) message);
            msg.addProperties(new ArrayList<>(parsed.keySet()));
            return Mono.just(msg);
            //判断物消息为修改属性消息，将转换结果集合key作为需要修改属性的集合
        } else if (message instanceof WriteThingPropertyMessage) {
            WriteThingPropertyMessage<?> msg = ((WriteThingPropertyMessage<?>) message);
            parsed.forEach(msg::addProperty);
            return Mono.just(msg);
            //判断物消息为读取属性消息回复，将转换结果作为读取属性的回复
        } else if (message instanceof ReadThingPropertyMessageReply) {
            ReadThingPropertyMessageReply msg = ((ReadThingPropertyMessageReply) message);
            msg.success(parsed);
            return Mono.just(msg);
            //判断物消息为修改属性消息回复，将转换结果作为修改属性的回复
        } else if (message instanceof WriteThingPropertyMessageReply) {
            WriteThingPropertyMessageReply msg = ((WriteThingPropertyMessageReply) message);
            msg.success(parsed);
            return Mono.just(msg);
        }
        throw new UnsupportedOperationException("unsupported message type" + message.getMessageType());
    }

    /**
     * 做信息转换
     *
     * @param expr 需要转换的信息
     * @param context 上下文
     * @return 转换后的信息
     */
    private Map<String, Object> doConvert(Map<String, Object> expr, Map<String, Object> context) {
        Map<String, Object> parsed = Maps.newLinkedHashMapWithExpectedSize(expr.size());
        for (Map.Entry<String, Object> entry : expr.entrySet()) {
            String key = (String) convert(entry.getKey(), context);
            //忽略不需要处理的数据
            if (StringUtils.isEmpty(key) || isIgnore(key)) {
                continue;
            }
            //需要转换的数据为迭代器类型时，遍历处理转换数据的子元素数据
            Object value = entry.getValue();
            if (value instanceof Iterable) {
                for (Object node : ((Iterable<?>) value)) {
                    Object nodeValue = convert(node, context);
                    if (!StringUtils.isEmpty(nodeValue)) {
                        parsed.put(key, nodeValue);
                    }
                }
            }
        }
        return parsed;
    }

    protected Object convert(Object value, Map<String, Object> context) {
        return value;
    }

    /**
     * 获取物消息中需要转换的信息，如属性上报消息中的属性，功能调用消息中的输入参数
     *
     * @param message 物消息
     * @return 物消息中需要转换的信息
     */
    private Map<String, Object> getNeedConvertInfo(ThingMessage message) {
        Map<String, Object> expr = null;
        if (message instanceof PropertyMessage) {
            expr = ((PropertyMessage) message).getProperties();
        } else if (message instanceof ThingFunctionInvokeMessage) {
            expr = ((ThingFunctionInvokeMessage<?>) message).inputsToMap();
        } else if (message instanceof ReadThingPropertyMessage) {
            expr = ((ReadThingPropertyMessage<?>) message)
                .getProperties()
                .stream()
                .collect(Collectors.toMap(Function.identity(), Function.identity()));
        } else if (message instanceof WriteThingPropertyMessage) {
            expr = ((WriteThingPropertyMessage<?>) message).getProperties();
        }
        Map<String, Object> all = Maps.newLinkedHashMapWithExpectedSize(32);
        all.putAll(getDefaultExpression());
        if (expr != null) {
            all.putAll(expr);
        }
        return all;
    }

    /**
     * 判断当前正在处理标识是否需要忽略处理
     *
     * @param key 标识
     * @return 是否需要忽略处理的结果
     */
    protected boolean isIgnore(String key) {
        if ((!CollectionUtils.isEmpty(includes) && !includes.contains(key))) {
            return true;
        }
        return !CollectionUtils.isEmpty(excludes) && excludes.contains(key);
    }

    protected Map<String, ?> getDefaultExpression() {
        return Collections.emptyMap();
    }
}
